package com.ahcloud.edas.gatewaylog.biz.infrastructure.log;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.ahcloud.edas.gatewaylog.biz.domain.log.dto.GatewayRequestLogDTO;
import com.ahcloud.edas.gatewaylog.biz.infrastructure.configuration.OperateLogProperties;
import com.ahcloud.edas.gatewaylog.biz.infrastructure.log.client.LogConsumeClient;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @program: ahcloud-operatelog
 * @description:
 * @author: YuKai Fan
 * @create: 2024/5/26 20:57
 **/
@Slf4j
public abstract class AbstractLogCollector<T extends LogConsumeClient<L>, L extends GatewayRequestLogDTO> implements LogCollector<L> {

    private final int bufferSize;
    private final BlockingQueue<L> bufferQueue;

    private long lastPushTime;

    private final AtomicBoolean started = new AtomicBoolean(true);

    protected AbstractLogCollector(OperateLogProperties properties) {
        this.bufferSize = properties.getBufferQueueSize();
        this.bufferQueue = new LinkedBlockingQueue<>(bufferSize);
    }

    @Override
    public void start() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                200, Integer.MAX_VALUE,60000L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(Integer.MAX_VALUE),
                ThreadFactoryBuilder.create().setNamePrefix("operatelog-shared").setDaemon(true).build(),
                new ThreadPoolExecutor.AbortPolicy()
        );
        started.set(true);
        threadPoolExecutor.execute(this::consume);
    }

    @Override
    public void collect(L log) {
        if (Objects.isNull(log) || Objects.isNull(getLogConsumeClient())) {
            return;
        }
        if (bufferQueue.size() < bufferSize) {
            bufferQueue.add(log);
        }
    }

    @Override
    public void close() throws Exception {
        started.set(false);
        T client = getLogConsumeClient();
        if (client != null) {
            client.close();
        }
    }

    /**
     * 获取日志消费客户端
     *
     * @return log consume client
     */
    protected abstract T getLogConsumeClient();


    private void consume() {
        while (started.get()) {
            int diffTimeMSForPush = 1000;
            try {
                List<L> logs = Lists.newArrayList();
                int size = bufferQueue.size();
                long time = System.currentTimeMillis();
                long timeDiffMs = time - lastPushTime;
                int batchSize = 100;
                if (size >= batchSize || timeDiffMs > diffTimeMSForPush) {
                    bufferQueue.drainTo(logs, batchSize);
                    T client = getLogConsumeClient();
                    if (Objects.nonNull(client)) {
                        client.consume(logs);
                    }
                    lastPushTime = time;
                } else {
                    TimeUnit.MILLISECONDS.sleep(diffTimeMSForPush);
                }
            } catch (Exception ignored) {
                log.error("DefaultLogCollector collect log error", ignored);
                try {
                    TimeUnit.MILLISECONDS.sleep(diffTimeMSForPush);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
